package com.yangye.springclouddemomqtt.config;

import com.yangye.springclouddemomqtt.utils.RandomUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * @author yangye
 * @date 2019/11/4 0004
 */
@Slf4j
@IntegrationComponentScan
@Configuration
public class MqttConfig {

	@Autowired
	private MqttProperty mqttProperty;

	@Bean
	public MqttConnectOptions getMqttConnectOptions() {
		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
		mqttConnectOptions.setUserName(mqttProperty.getUsername());
		mqttConnectOptions.setKeepAliveInterval(2);
		mqttConnectOptions.setMaxInflight(100000000);
		mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
		mqttConnectOptions.setServerURIs(new String[]{mqttProperty.getHost()});
		mqttConnectOptions.setConnectionTimeout(60);
		return mqttConnectOptions;
	}

	@Bean
	public MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		factory.setConnectionOptions(getMqttConnectOptions());
		return factory;
	}

	@Bean
	public MessageChannel mqttInputChannel() {
		return new DirectChannel();
	}

	@Bean
	public MessageChannel mqttOutboundChannel() {
		return new DirectChannel();
	}


	@Bean
	@ServiceActivator(inputChannel = "mqttOutboundChannel")
	public MessageHandler mqttOutbound() {
		//clientId使用随机数产生
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperty.getClientId() + "_out_" + RandomUtil.randomString(6), mqttClientFactory());
		messageHandler.setAsync(true);
		messageHandler.setDefaultTopic(mqttProperty.getDefaultTopic());
		messageHandler.setDefaultRetained(false);
		return messageHandler;
	}


	@Bean
	public MessageProducer inbound() {
		String[] inboundTopics = mqttProperty.getDefaultTopic().split(",");
		MqttPahoMessageDrivenChannelAdapter adapter =
				new MqttPahoMessageDrivenChannelAdapter(mqttProperty.getClientId() + "_in_" + RandomUtil.randomString(6), mqttClientFactory(), inboundTopics);
		adapter.setCompletionTimeout(5000);
		adapter.addTopic("xxx");
		adapter.setConverter(new DefaultPahoMessageConverter());
		adapter.setQos(1);
		adapter.setOutputChannel(mqttInputChannel());
		return adapter;
	}


	@Bean
	@ServiceActivator(inputChannel = "mqttInputChannel")
	public MessageHandler handler() {
		return message ->
				log.info("主题：{}，消息接收到的数据：{}", message.getHeaders().get("mqtt_receivedTopic"), message.getPayload());
	}
}
